package com.chengqs.dfs.namenode.server;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.chengqs.dfs.namenode.rpc.model.*;
import com.chengqs.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.stub.StreamObserver;
import javafx.scene.shape.Path;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;

/**
 * <B>主类名称：</B>NameNodeRpcServer<BR>
 * <B>概要说明：</B>NameNode的rpc服务的接口.<BR>
 *
 * @author: chengqs
 * @create: 2025-03-30 03:49:45
 **/
@Slf4j
public class NameNodeServiceImpl implements NameNodeServiceGrpc.NameNodeService {

    public static final Integer STATUS_SUCCESS = 1;
    public static final Integer STATUS_FAILURE = 2;
    public static final Integer STATUS_SHUTDOWN = 3;
    public static final Integer STATUS_DUPLICATE = 4;

    public static final Integer BACKUP_NODE_FETCH_SIZE = 10;

    /**
     * 负责管理元数据的核心组件
     */
    private FSNameSystem nameSystem;
    /**
     * 负责管理集群中所有的datanode的组件
     */
    private DataNodeManager datanodeManager;

    // 是否还在运行
    private volatile Boolean isRunning = true;

    // 当前backup节点同步到了哪一条txid了
    // private long syncedTxid = 0;

    // 当前缓冲的一小部分editslog
    private JSONArray currentBufferedEditsLog = new JSONArray();

    // 当前内存里缓冲了哪个磁盘文件的数据
    private String bufferedFlushedTxid;

    // 当前缓存里editslog最大的一个txid
    private long currentBufferedMaxTxid = 0L;

    public NameNodeServiceImpl(
            FSNameSystem nameSystem,
            DataNodeManager datanodeManager) {
        this.nameSystem = nameSystem;
        this.datanodeManager = datanodeManager;
    }

    /**
     * DataNode进行注册
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void register(RegisterRequest request, StreamObserver<RegisterResponse> responseObserver) {
        datanodeManager.register(request.getIp(), request.getHostname());
        RegisterResponse response = RegisterResponse.newBuilder()
                .setStatus(STATUS_SUCCESS)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * DataNode进行心跳
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void heartbeat(HeartbeatRequest request, StreamObserver<HeartbeatResponse> responseObserver) {
        datanodeManager.heartbeat(request.getIp(), request.getHostname());
        HeartbeatResponse response = HeartbeatResponse.newBuilder()
                .setStatus(STATUS_SUCCESS)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 创建目录
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void mkdir(MkdirRequest request, StreamObserver<MkdirResponse> responseObserver) {
        try {
            MkdirResponse response = null;
            if (!this.isRunning) {
                response = MkdirResponse.newBuilder()
                        .setStatus(STATUS_SHUTDOWN)
                        .build();
            } else {
                this.nameSystem.mkdir(request.getPath());

                // log.info("创建目录：path = {}", request.getPath());

                response = MkdirResponse.newBuilder()
                        .setStatus(STATUS_SUCCESS)
                        .build();
            }
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭NameNode
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void shutdown(ShutdownRequest request, StreamObserver<ShutdownResponse> responseObserver) {
        this.isRunning = false;
        this.nameSystem.flush();

        nameSystem.saveCheckpointTxid();

        log.info("优雅关闭namenode......");

        ShutdownResponse response = ShutdownResponse.newBuilder()
                .setStatus(STATUS_SUCCESS)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 拉取editslog
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void fetchEditsLog(FetchEditsLogRequest request, StreamObserver<FetchEditsLogResponse> responseObserver) {
        if (!isRunning) {
            FetchEditsLogResponse response = FetchEditsLogResponse.newBuilder()
                    .setEditsLog(new JSONArray().toJSONString())
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            return;
        }

        long syncedTxid = request.getSyncedTxid();

        FetchEditsLogResponse response = null;
        JSONArray fetchedEditsLog = new JSONArray();

        List<String> flushedTxids = nameSystem.getEditLog().getFlushedTxids();

        // 如果此时还没有刷出来任何磁盘文件的话，那么此时数据仅仅存在于内存缓冲里
        if (flushedTxids.size() == 0) {
            log.info("暂时没有任何磁盘文件，直接从内存缓冲中拉取......");
            fetchFromBufferedEditsLog(syncedTxid, fetchedEditsLog);
        }
        // 如果此时发现已经有落地磁盘的文件了，这个时候就要扫描所有的磁盘文件的索引范围
        else {
            // 第一种情况，你要拉取的txid是在磁盘文件里的
            // 有磁盘文件，而且内存里还缓存了某个磁盘文件的数据了
            if (bufferedFlushedTxid != null) {
                // 如果要拉取的数据就在当前缓存的磁盘文件的数据里
                if (existInFlushedFile(syncedTxid, bufferedFlushedTxid)) {
                    log.info("上一次已经缓存过磁盘文件的数据，直接从磁盘文件缓存中拉取editslog......");
                    fetchFromCurrentBuffer(syncedTxid, fetchedEditsLog);
                }
                // 如果要拉取的数据不在当前缓存的磁盘文件的数据里，需要从下一个磁盘文件去拉取
                else {
                    String nextFlushedTxid = getNextFlushedTxid(flushedTxids, bufferedFlushedTxid);
                    // 如果可以找到下一个磁盘文件，那么就从下一个磁盘文件里开始读取数据
                    if (nextFlushedTxid != null) {
                        log.info("上一次缓存的磁盘文件找不到要拉取的数据，从下一个磁盘文件中拉取editslog......");
                        fetchFromFlushedFile(syncedTxid, nextFlushedTxid, fetchedEditsLog);
                    }
                    // 如果没有找到下一个文件，此时就需要从内存里去继续读取
                    else {
                        log.info("上一次缓存的磁盘文件找不到要拉取的数据，而且没有下一个磁盘文件，尝试从内存缓冲中拉取editslog......");
                        fetchFromBufferedEditsLog(syncedTxid, fetchedEditsLog);
                    }
                }
            }
            // 第一次尝试从磁盘文件里去拉取数据
            else {
                // 遍历所有的磁盘文件的索引范围，0~390，391~792
                boolean fetchedFromFlushedFile = false;
                for (String flushedTxid : flushedTxids) {
                    if (existInFlushedFile(syncedTxid, flushedTxid)) {
                        log.info("尝试从磁盘文件中拉取editslog，flushedTxid=" + flushedTxid);
                        // 此时可以把这个磁盘文件以及下一个磁盘文件里的数据读取出来，放到内存里来缓存
                        // 就怕一个磁盘文件的数据不足10条
                        fetchFromFlushedFile(syncedTxid, flushedTxid, fetchedEditsLog);
                        fetchedFromFlushedFile = true;
                        break;
                    }
                }
                // 第二种情况，你要拉取的txid已经比磁盘文件的txid都大，此时你需要从内存缓冲里取数据
                // 如果没有找到下一个文件，此时就需要从内存里去继续读取
                if (!fetchedFromFlushedFile) {
                    log.info("所有磁盘文件都没找到要拉取的editslog，尝试直接从内存缓冲中拉取editslog......");
                    fetchFromBufferedEditsLog(syncedTxid, fetchedEditsLog);
                }
            }
        }
        response = FetchEditsLogResponse.newBuilder()
                .setEditsLog(fetchedEditsLog.toJSONString())
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 更新checkpoint的txid
     * @param request
     * @param responseObserver
     */
    @Override
    public void updateCheckpointTxid(UpdateCheckpointTxidRequest request, StreamObserver<UpdateCheckpointTxidResponse> responseObserver) {
        long txid = request.getTxid();
        nameSystem.setCheckpointTxid(txid);

        UpdateCheckpointTxidResponse response = UpdateCheckpointTxidResponse.newBuilder()
                .setStatus(1)
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();

    }

    /**
     * 创建文件
     * @param request
     * @param responseObserver
     */
    @Override
    public void create(CreateFileRequest request, StreamObserver<CreateFileResponse> responseObserver) {
        // 把文件名的查重和创建文件夹放在一起执行
        // 如果说很多个客户端万一同时要发起文件创建，都有一个文件名过来
        // 多线程并发的情况下，文件名的查重和创建都是正确执行的
        // 就必须得在同步的代码块执行这个功能逻辑
        try {
            CreateFileResponse response = null;
            if (!this.isRunning) {
                response = CreateFileResponse.newBuilder()
                        .setStatus(STATUS_SHUTDOWN)
                        .build();
            } else {
                String filename = request.getFilename();
                Boolean success = nameSystem.create(filename);

                if (success) {
                    response = CreateFileResponse.newBuilder()
                            .setStatus(STATUS_SUCCESS)
                            .build();
                } else {
                    response = CreateFileResponse.newBuilder()
                            .setStatus(STATUS_DUPLICATE)
                            .build();
                }
            }
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 就是从内存缓冲的editslog中去拉取数据
     *
     * @param fetchedEditsLog
     */
    private void fetchFromBufferedEditsLog(long syncedTxid, JSONArray fetchedEditsLog) {
        // 如果要拉取的txid还在上一次内存缓存中，此时继续从内存缓存中来拉取即可
        long fetchTxid = syncedTxid + 1;
        if (fetchTxid <= currentBufferedMaxTxid) {
            log.info("尝试从内存缓冲拉取的时候，发现上一次内存缓存有数据可供拉取......");
            fetchFromCurrentBuffer(syncedTxid, fetchedEditsLog);
            return;
        }

        // 必须得重新把内存缓冲里的数据加载到内存缓存里来
        currentBufferedEditsLog.clear();

        String[] bufferedEditsLog = nameSystem.getEditLog().getBufferedEditsLog();

        if (bufferedEditsLog != null) {
            for (String editsLog : bufferedEditsLog) {
                currentBufferedEditsLog.add(JSONObject.parseObject(editsLog));
                currentBufferedMaxTxid = JSONObject.parseObject(editsLog).getLongValue("txid");
            }

            bufferedFlushedTxid = null;

            fetchFromCurrentBuffer(syncedTxid, fetchedEditsLog);
        }
    }

    /**
     * 从当前已经在内存里缓存的数据中拉取editslog
     * @param fetchedEditsLog
     */
    private void fetchFromCurrentBuffer(long syncedTxid, JSONArray fetchedEditsLog) {
        int fetchCount = 0;
        long fetchTxid = syncedTxid + 1;
        for (int i = 0; i < currentBufferedEditsLog.size(); i++) {
            if (currentBufferedEditsLog.getJSONObject(i).getLong("txid") == fetchTxid) {
                fetchedEditsLog.add(currentBufferedEditsLog.getJSONObject(i));
                fetchTxid = currentBufferedEditsLog.getJSONObject(i).getLong("txid") + 1;
                fetchCount++;
            }
            if (fetchCount == BACKUP_NODE_FETCH_SIZE) {
                break;
            }
        }
    }

    /**
     * 获取下一个磁盘文件对应的txid范围
     * @param flushedTxids
     * @param bufferedFlushedTxid
     * @return
     */
    private String getNextFlushedTxid(List<String> flushedTxids, String bufferedFlushedTxid) {
        for (int i = 0; i < flushedTxids.size(); i++) {
            if (flushedTxids.get(i).equals(bufferedFlushedTxid)) {
                if (i + 1 < flushedTxids.size()) {
                    return flushedTxids.get(i + 1);
                }
            }
        }
        return null;
    }

    /**
     * 判断这个磁盘文件是否已经刷盘了
     * @param flushedTxid
     * @return
     */
    private boolean existInFlushedFile(long syncedTxid, String flushedTxid) {
        String[] flushedTxidSplit = flushedTxid.split("_");

        long startTxid = Long.valueOf(flushedTxidSplit[0]);
        long endTxid = Long.valueOf(flushedTxidSplit[1]);
        long fetchTxid = syncedTxid + 1;

        if (fetchTxid >= startTxid && fetchTxid <= endTxid) {
            return true;
        }
        return false;
    }

    /**
     * 从已经刷入磁盘的文件里读取editsLog，同时缓存这个文件数据到内存
     * @param flushedTxid
     */
    private void fetchFromFlushedFile(long syncedTxid, String flushedTxid, JSONArray fetchedEditsLog) {
        try {

            String[] flushedTxidSplit = flushedTxid.split("_");

            long startTxid = Long.valueOf(flushedTxidSplit[0]);
            long endTxid = Long.valueOf(flushedTxidSplit[1]);

            String currentEditsLogFile = "D:\\java\\儒猿课堂\\儒猿海量小文件分布式存储项目\\edits-" +
                    startTxid + "-" + endTxid + ".log";
            List<String> editsLogs = Files.readAllLines(Paths.get(currentEditsLogFile), StandardCharsets.UTF_8);

            currentBufferedEditsLog.clear();
            for (String editsLog : editsLogs) {
                currentBufferedEditsLog.add(JSONObject.parseObject(editsLog));
                currentBufferedMaxTxid = JSONObject.parseObject(editsLog).getLongValue("txid");
            }

            bufferedFlushedTxid = flushedTxid; // 缓存了某个刷入磁盘文件的数据

            fetchFromCurrentBuffer(syncedTxid, fetchedEditsLog);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
